/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.io.File;
import java.nio.ByteBuffer;

import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.storage.columns.IColumnFactory;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;

public class CabinetLoader {

  private final IBufferCache bufferCache;
  private final ILexiconLoader lexiconBuilder;
  protected final int cabinetFileId;

  private int nextPage = 0;

  public CabinetLoader(final IBufferCache bufferCache, final String cabFileName,
                       final IFileMapProvider fileMapProvider,
                       final ILexiconLoader lexLoader) throws HyracksDataException {
    // Assign injected dependencies.
    this.bufferCache = bufferCache;
    this.lexiconBuilder = lexLoader;

    // Initialize cabinet file inside buffer cache.
    FileReference file = new FileReference(new File(cabFileName));
    bufferCache.createFile(file);
    this.cabinetFileId = fileMapProvider.lookupFileId(file);
    bufferCache.openFile(this.cabinetFileId);
  }

  public void close() {
    this.lexiconBuilder.close();
  }

  public int closeColumn(final ColumnLoader<?> column) {
    String key = column.getKey();
    int columnAddress = column.fileColumn();
    this.lexiconBuilder.addKey(key, columnAddress);
    column.close();
    return columnAddress;
  }

  public <VT extends IValueType> ColumnLoader<VT>
      startColumn(final String columnKey, final IColumnFactory<VT> columnFactory) {
    ColumnLoader<VT> colLoader = columnFactory.makeColumnLoader(this, columnKey);
    colLoader.open();
    return colLoader;
  }

  public int writeDrawer(final byte[] array) {
    long dpid = BufferedFileHandle.getDiskPageId(this.cabinetFileId, this.nextPage);
    try {
      ICachedPage page = this.bufferCache.pin(dpid, true);
      ByteBuffer pageBuffer = page.getBuffer();
      page.acquireWriteLatch();
      System.arraycopy(array, 0, pageBuffer.array(), 0, ColumnStore.BYTES_IN_DRAWER);
      page.releaseWriteLatch();
      this.bufferCache.unpin(page);
    } catch (HyracksDataException e) {
      throw new RuntimeException(e);
    }
    return this.nextPage++;
  }

}
